package graph;

import java.io.File;
import java.util.Collection;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

import mapreduce.IntegerFormat;
import mapreduce.Record;

public class DirectedEdges {
	public DirectedEdges(EdgesImplFactory implFactory, String edgesName) {
		m_directEdgesImpl = implFactory.getImpl(edgesName + "_direct");
		m_inverseEdgesImpl = implFactory.getImpl(edgesName + "_inverse");
	}

	EdgesImpl m_directEdgesImpl;
	EdgesImpl m_inverseEdgesImpl;

	static enum NodeType {
		SRC, DES
	}

	public void load(File directFile, File inverseFile) {
		m_directEdgesImpl.load(directFile);
		m_inverseEdgesImpl.load(inverseFile);
	}

	public Set<Edge> getEdgesFrom(int srcId) {
		EdgesImpl.IdWeight[] weighs = m_directEdgesImpl.getEdgesFrom(srcId);
		return collectEdges(srcId, weighs, NodeType.SRC);
	}

	public Set<Edge> getEdgesTo(int desId) {
		EdgesImpl.IdWeight[] weights = m_inverseEdgesImpl.getEdgesFrom(desId);
		return collectEdges(desId, weights, NodeType.DES);
	}

	private Set<Edge> collectEdges(int nodeId, EdgesImpl.IdWeight[] weights, NodeType nodeType){
		Set<Edge> edges = new HashSet<Edge>();
		if(weights == null){
			return edges;
		}
		int num = weights.length;
		for(int i = 0 ; i < num ; i ++){
			int id1 = nodeId;
			int id2 = weights[i].m_id;
			double weight = weights[i].m_weight;
			if(nodeType == NodeType.SRC){
				Edge edge = new Edge(id1, id2, weight);
				edges.add(edge);
			}
			else{
				Edge edge = new Edge(id2, id1, weight);
				edges.add(edge);
			}
		}
		return edges;
		
	}
	
	public void addEdges(Collection<Edge> edges){
		Iterator<Edge> edgeIt = edges.iterator();
		while(edgeIt.hasNext()){
			Edge edge = edgeIt.next();
			addEdge(edge);
		}
	}

	public void addEdge(Edge edge) {
		int id1 = edge.m_srcId;
		int id2 = edge.m_desId;
		double weight = edge.m_weight;

		EdgesImpl.IdWeight[] idWeights1 = new EdgesImpl.IdWeight[1];
		idWeights1[0] = new EdgesImpl.IdWeight(id2, weight);
		m_directEdgesImpl.add(id1, idWeights1);

		EdgesImpl.IdWeight[] idWeights2 = new EdgesImpl.IdWeight[1];
		idWeights2[0] = new EdgesImpl.IdWeight(id1, weight);
		m_inverseEdgesImpl.add(id2, idWeights2);
	}
	
	public Set<Integer> getSrcNodes(){
		return m_directEdgesImpl.getSrcNodes();
	}
	
	public Set<Integer> getDesNodes(){
		return m_directEdgesImpl.getDesNodes();
	}

	public float getWeightSumFrom(int srcId) {
		Set<Edge> edges = getEdgesFrom(srcId);
		return sumWeight(edges);
	}

	public float getWeightSumTo(int desId) {
		Set<Edge> edges = getEdgesTo(desId);
		return sumWeight(edges);
	}

	protected float sumWeight(Collection<Edge> edges) {
		float weightSum = 0;
		Iterator<Edge> edgeIt = edges.iterator();
		while (edgeIt.hasNext()) {
			Edge edge = edgeIt.next();
			weightSum += edge.m_weight;
		}
		return weightSum;
	}
	
	static public void main(String args[]){
		EdgesImplFactory implFactory = new DbEdgesImplFactory("/home/czhai/jinghe/local/WORK/google-code/twitter-se/data/graph");
		String edgesName = "favorite";
		DirectedEdges favoriteEdges = new DirectedEdges(implFactory, edgesName);
		Set<Integer> tweetIds = favoriteEdges.getDesNodes();
		for(int tweetId: tweetIds){
			Set<Edge> edgeSet = favoriteEdges.getEdgesTo(tweetId);
			System.out.printf("%d:", tweetId);
			for(Edge edge: edgeSet){
				System.out.printf("%d;", edge.m_srcId);
			}
			System.out.println();
		}
	}
}
